#include <server.h>

#define STATE_IDLE				0
#define STATE_READY				1
#define STATE_COMMIT_YES		2
#define STATE_COMMIT_NO			3

int main(int argc, char** argv) {
	RF244BPacket*	pack;
	RFEvent			event;

	/* Initialize global variable */
	WBBufferSize = 0;
	fileDescriptor = -1;
	myState = STATE_IDLE;
	clientIP = 0;
	
	/* Initialize baseline time */
	struct timeval timenow;
	gettimeofday(&timenow, 0);
	TimeZero = (timenow.tv_sec * 1000 + timenow.tv_usec / 1000);
	
	/* Check parameters */
	if ((argc != 5) || (strcmp(argv[1], "-port") != 0) || (strcmp(argv[3], "-mount") != 0)) {
		RFError("Please input replFsServer -port <port#> -mount <filepath>");
	}
	
	myRFPort = atoi(argv[2]);
	strcpy(filePath, argv[4]);
	if (filePath[strlen(filePath)-1] != '/') {
		strcat(filePath, "/");
	}
	
	printf("Filepath = %s , port = %d\n", filePath, myRFPort);

	RFDebug("Creating directory");
	
	/* Create the directory */
	if (mkdir(filePath, S_IRWXU|S_IRGRP|S_IXGRP) != 0) {
		RFError("machine already in use");
	}
	
	RFDebug("Setting up network");
	
	/* Set up connection */
	netInit(myRFPort);
	
	RFDebug("Starting to listen for packet");
	
	/* Main loop listening for packets */
	while (1) {
		NextEvent(&event, mySocket, 500);
RFDebug("Next Event");
		/* Nothing happens */
		if (event.eventType != EVENT_NETWORK) {
			continue;		
		}
		pack = event.eventDetail;
		
		/* Check if the packet is intended for me */
		if ((pack->receiverIP != myIP) && (pack->receiverIP != 0)) {
RFWarn("IP does not match");
			continue;
		}
			
		/* Record IP of the client */
		if (clientIP == 0) {
			clientIP = pack->senderIP;
		}
		
		/* Check if the packet is from the client */
		if (clientIP != pack->senderIP) {
			RFWarn("Receive packet from sb other than client, discard");
			continue;
		}
		
		if ((fileDescriptor != -1) && (fileDescriptor != pack->fd)) {
			RFWarn("Receive packet intended for another file, discard");
			continue;
		}
		
		messageID = pack->messageID;
		
		switch (pack->type) {
			case (PACKET_OPEN) : {
				POPEN* packX = (POPEN *) &(pack->body);
				openFile (pack->fd, packX->name);
				break;
			}
			case (PACKET_WB) : {
				PWriteBlock* packX = (PWriteBlock *) &(pack->body);
				packX->blockSize = ntohl(packX->blockSize);
				packX->byteOffset = ntohl(packX->byteOffset);
				writeBlock (pack->messageID, packX);
				break;
			}
			case (PACKET_COMMIT) : {
				PCommit* packX = (PCommit *) &(pack->body);
				packX->num = ntohl(packX->num);
				commitTry(packX->num);
				break;
			}
			case (PACKET_COMMITNOW) : {
				commitFile();
				break;
			}
			case (PACKET_ABORT) : {
				abortFile();
				break;
			}
			case (PACKET_CLOSE) : {
				closeFile();
				break;
			}
			default : {
				RFWarn("Unrecognized packet type");
			}		
		}	
		
	}
	
	return 0;
}

/* Open a file on server */
void openFile (int fd, char *fileName) {
	fileDescriptor = fd;
	
	/* Check the state */
	if (myState != STATE_IDLE) {
		RFWarn("Out of order packet");
		sendACK();
		return;
	}
	/* Open file */
	memset(myFileName, 0, sizeof (myFileName));
	strcat(myFileName, filePath);
	strcat(myFileName, fileName);
	myFileFd = open( myFileName, O_RDWR|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR );
	if ( myFileFd < 0 ) {
		RFWarn("Open file failed at server");
		printf("File name = %s\n", myFileName);		//Debug
		sendNACK();
		return;
	}

	printf("Open file succeeded. File name = %s\n", myFileName);

	sendACK();
	myState = STATE_READY;
	
	mkdir("./tmp", S_IRWXU|S_IRGRP|S_IXGRP);
	/* Try to create a copy of the file */
	strcpy(tmpFileName, "./tmp/CS244BRF.XXXXXX");
	if (mkstemp(tmpFileName) < 0) {
		RFWarn("Cannot create tmp file!");
	}
	
	/* Open the temp file */
	tmpFileFd = open( tmpFileName, O_RDWR|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR );
	if (tmpFileFd < 0) {
		RFWarn("Cannot open tmp file!");
	}	
printf("tmpFileName : %s\n", tmpFileName);	
	
	/* Initialize file related resources */
	WBBufferSize = 0;
	for (int i = 0; i < MaxWriteBlock; i++) {
		WBBuffer[i].blockSize = -1;
	}
	return;
}




/* Handle write block request, buffer them */
void writeBlock (int idx, PWriteBlock *packX) {
	/* Check the state */
	if (myState != STATE_READY) {
		RFDebug("Out of order packet in writeBlock");		
		return;
	}
	
printf("WB: blockSize = %d, offset = %d, content = %s\n", packX->blockSize, packX->byteOffset, packX->buffer);		
	WBBuffer[idx].blockSize = packX->blockSize;
	WBBuffer[idx].byteOffset = packX->byteOffset;
	memcpy(WBBuffer[idx].buffer, packX->buffer, packX->blockSize);
	return;
}

/* First phase of 2PC */
void commitTry(int size) {
	RF244BPacket*	pack;
	RFEvent			event;

	/* Check the state */
	if (myState != STATE_READY) {
		RFDebug("Out of order packet in commit try");		
		if (myState == STATE_COMMIT_YES)
			sendACK();
		else
			sendNACK();
		return;
	}
	
	/* Collect missing write block pieces */
	int tries = 0;
	int timestamp = getCurrentTime() + TIMEOUT;
	WBBufferSize = size;
	
	bool hasDone = TRUE;
	for (int i = 0; i < size; i++) {
		if (WBBuffer[i].blockSize < 0) {
printf("Data %d is missing, sending request\n", i);
			sendReqData(i);
			hasDone = FALSE;
		}
	}
	
	int nextTimeOut;
	while (!hasDone) {
		nextTimeOut = timestamp;
		
		NextEvent(&event, mySocket, nextTimeOut - getCurrentTime());
		pack = event.eventDetail;
		if (event.eventType == EVENT_NETWORK) {			
			/* Check receiver IP */
			if ((pack->receiverIP != myIP) && (pack->receiverIP != 0)) {
				printf("This packet is not for me\n");
				continue;
			}
			
			/* If abort occurs, other server has problems, abort */
			if (pack->type == PACKET_ABORT) {
				messageID = pack->messageID;
				abortFile();
				return;
			}
			
			/* We are waiting for WB messages */
			if (pack->type != PACKET_WB) {
				continue;
			}
			
			int idx = pack->messageID;
			PWriteBlock *packX = (PWriteBlock *) &(pack->body);
			packX->blockSize = ntohl(packX->blockSize);
			packX->byteOffset = ntohl(packX->byteOffset);
			writeBlock (idx, packX);
			timestamp = getCurrentTime() + TIMEOUT;			
		}
		
		if (timestamp <= getCurrentTime()) {
			tries++;
			if (tries > MAX_TRY) {
				RFWarn("Client is down on CommitTry");
				break;
			}
			for (int i = 0; i < size; i++) {
				if (WBBuffer[i].blockSize < 0) {
printf("Data %d is missing, sending request\n", i);				
					sendReqData(i);
				}
			}
			timestamp = getCurrentTime() + TIMEOUT;
		}
		
		/* Check if done successfully */
		hasDone = TRUE;
		for (int i = 0; i < size; i++) 
			if (WBBuffer[i].blockSize < 0) {
				hasDone = FALSE;
				break;
			}
	}	
	
	if (!hasDone) {
		sendNACK();
		myState = STATE_COMMIT_NO;    		
		return;
	}
	
	/* Write to the temp file */
	for (int i = 0; i < WBBufferSize; i++) {
		if ( lseek( tmpFileFd, WBBuffer[i].byteOffset, SEEK_SET ) < 0 ) {
    		RFWarn("Seek error during commit try!");
    		sendNACK();
    		myState = STATE_COMMIT_NO;    		
			close(tmpFileFd);    		
    		return;
  		}

		if ( write( tmpFileFd, WBBuffer[i].buffer, WBBuffer[i].blockSize ) < 0 ) {
    		RFWarn( "WriteBlock write error during commit try" );
    		sendNACK();
			myState = STATE_COMMIT_NO;   
			close(tmpFileFd); 	
	    	return;
	  	}
	}

	/* Commit try is successful */
	RFDebug("Commit try succeeded");

	sendACK();
	myState = STATE_COMMIT_YES;
}

/* Second phase of the 2PC */
void commitFile() {
	/* Check the state */
	if (myState != STATE_COMMIT_YES) {
		RFWarn("Out of order packet in commitFile");
		sendACK();
		return;
	}
	
	/* Do the real commit */
	int bytes;
	char line[512];
	lseek (tmpFileFd, 0, SEEK_SET);
	lseek (myFileFd, 0, SEEK_SET);
	while((bytes = read(tmpFileFd, line, sizeof(line))) > 0) {
    	write(myFileFd, line, bytes);
printf("Bytes written : %d\n", bytes);
    }
	
printf("Commit Now succeeded\n");
	
	sendACK();
	myState = STATE_READY;
	
	/* Initialize some parameter for next use */
	WBBufferSize = 0;
	for (int i = 0; i < MaxWriteBlock; i++)
		WBBuffer[i].blockSize = -1;
	
	return;
}

/* Abort all changes since last abort/commit */
void abortFile () {
	/* Check the state */
	if (myState == STATE_IDLE) {
		RFWarn("Out of order packet in abort file");
		sendACK();
		return;
	}

	sendACK();	
	myState = STATE_READY;
	
	/* Initialize some parameter for next use */
	WBBufferSize = 0;
	for (int i = 0; i < MaxWriteBlock; i++)
		WBBuffer[i].blockSize = -1;

	printf("Rolling back\n");
	/* Roll back temporary file */
	int bytes;
	char line[512];
	tmpFileFd = open( tmpFileName, O_RDWR|O_CREAT|O_TRUNC, S_IRUSR|S_IWUSR );
	lseek (tmpFileFd, 0, SEEK_SET);
	lseek (myFileFd, 0, SEEK_SET);
	while((bytes = read(myFileFd, line, sizeof(line))) > 0) {
    	write(tmpFileFd, line, bytes);
printf("Bytes written : %d\n", bytes);    	
    }
	
	
	return;
}

/* Close the file */
void closeFile() {
	/* Check the state */
	if (myState != STATE_READY) {
		RFWarn("Out of order packet in closeFile");
		sendACK();
		return;
	}

	myState = STATE_IDLE;
	
	if ( close(fileDescriptor) < 0 ) {
		sendNACK();
	} else {
		sendACK();
	}
	
	/* Relinquish all control */
	if (fileDescriptor != -1) {
		fileDescriptor = -1;
	}

    /* Remove the temp files */
	if (remove (tmpFileName) < 0) {
		RFWarn("Cannot delete temp file");
	}	
	
printf("Close file succeeded\n");	
	return;	
}

/* Send acknowledgement message */
void sendACK() {
	RF244BPacket *pack = new RF244BPacket;
	
	pack->messageID = messageID;
	pack->type = PACKET_ACK;
	pack->receiverIP = clientIP;
	pack->fd = fileDescriptor;
	
	sendPacket(pack);
	free(pack);	
}

/* Send negative ack message */
void sendNACK() {
	RF244BPacket *pack = new RF244BPacket;

	pack->messageID = messageID;	
	pack->type = PACKET_NACK;
	pack->receiverIP = clientIP;
	pack->fd = fileDescriptor;
	
	sendPacket(pack);
	free(pack);	
}

/* Send data request */
void sendReqData(int idx) {
	RF244BPacket *pack = new RF244BPacket;
	
	pack->messageID = idx;
	pack->type = PACKET_REQDATA;
	pack->receiverIP = clientIP;
	pack->fd = fileDescriptor;
	
	sendPacket(pack);
	free(pack);	
}

/* Send the packet */
void sendPacket (RF244BPacket *pack) {

	pack->senderIP = myIP;
	pack->hash = hashFunc(pack);
	
	pack->messageID = htonl(pack->messageID);
	pack->senderIP = htonl(pack->senderIP);
	pack->receiverIP = htonl(pack->receiverIP);
	pack->type = htonl(pack->type);
	pack->fd = htonl(pack->fd);
	pack->hash = htonl(pack->hash);		

	if (sendto((int)mySocket, pack, sizeof(*pack), 0,
		   (struct sockaddr*) &groupAddr, sizeof(Sockaddr)) < 0)
	  { RFError("Send packet Error"); }	
}

/* Convert incoming packet */
void ConvertIncoming (RF244BPacket *pack) {
	pack->messageID = ntohl(pack->messageID);
	pack->senderIP	= ntohl(pack->senderIP);
	pack->receiverIP = ntohl(pack->receiverIP);
	pack->type = ntohl(pack->type);
	pack->fd = ntohl(pack->fd);
	pack->hash = ntohl(pack->hash);
	if (pack->hash != hashFunc(pack)) {
		RFWarn("Checksum failed, the packet is discarded");
		pack->receiverIP = 1;
		return;
	}
printf("Packet received, type = %d, fd = %d\n", pack->type, pack->fd);	
}

/* Hash function for checksum */
unsigned int hashFunc(RF244BPacket *pack) {
	unsigned int value = 0;
	for (int i = 0; i < 256; i++)
		value = value ^ pack->body[i];
	value = value ^ pack->messageID ^ pack->senderIP ^ pack->receiverIP ^ pack->type ^ pack->fd;
	return value;
}



/* ------------------------------------------------------------------ */
/* Initialize the connection, set up network */
void netInit(unsigned short portNum)
{
	Sockaddr	nullAddr;
	Sockaddr	*thisHost;
	char		buf[128];
	int			reuse;
	u_char      ttl;
	struct ip_mreq  mreq;

	myRFPort = htons(portNum);

	gethostname(buf, sizeof(buf));	
	if ((thisHost = resolveHost(buf)) == (Sockaddr *) NULL)
		RFError("who am I?");

	myAddr = (Sockaddr*) malloc(sizeof(Sockaddr));
	bcopy((caddr_t) thisHost, (caddr_t) (myAddr), sizeof(Sockaddr));
	
	/* Record my Ip address and port number */
	myIP = thisHost->sin_addr.s_addr;
	mySocket = socket(AF_INET, SOCK_DGRAM, 0);
	if (mySocket < 0)
		RFError("can't get socket");

printf("My IP is %ld, socket is %d\n", myIP, mySocket);

	/* SO_REUSEADDR allows more than one binding to the same
	   socket */
	reuse = 1;
	if (setsockopt(mySocket, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) < 0) {
		RFError("setsockopt failed (SO_REUSEADDR)");
	}

	nullAddr.sin_family = AF_INET;
	nullAddr.sin_addr.s_addr = htonl(INADDR_ANY);
	nullAddr.sin_port = myRFPort;
	if (bind(mySocket, (struct sockaddr *)&nullAddr, sizeof(nullAddr)) < 0)
		RFError("netInit binding");

	/* Multicast TTL:
	   0 restricted to the same host
	   1 restricted to the same subnet
	   32 restricted to the same site
	   64 restricted to the same region
	   128 restricted to the same continent
	   255 unrestricted

	   DO NOT use a value > 32. If possible, use a value of 1 when
	   testing.
	*/

	ttl = 32;
	if (setsockopt(mySocket, IPPROTO_IP, IP_MULTICAST_TTL, &ttl, sizeof(ttl)) < 0) {
		RFError("setsockopt failed (IP_MULTICAST_TTL)");
	}
	
	/* Join the multicast group */
	mreq.imr_multiaddr.s_addr = htonl(RFGROUP);
	mreq.imr_interface.s_addr = htonl(INADDR_ANY);
	if (setsockopt(mySocket, IPPROTO_IP, IP_ADD_MEMBERSHIP, (char *) &mreq, sizeof(mreq)) < 0) {
		RFError("setsockopt failed (IP_ADD_MEMBERSHIP)");
	}

	/* Get the multi-cast address ready to use in SendData()
           calls. */
	memcpy(&groupAddr, &nullAddr, sizeof(Sockaddr));
	groupAddr.sin_addr.s_addr = htonl(RFGROUP);
}

/* Get the next packet */
void NextEvent(RFEvent *event, int socket, int time)
{
	fd_set	fdmask;
	struct timeval timeout;
	int		ret;

    if (time < 0)
    	time = 0;
    	
  	while (1) 
  	{
	  	FD_ZERO(&fdmask);
	  	FD_SET(socket, &fdmask);
	  	for (ret = 0; ret <= 0;) 
	  	{
	      	timeout.tv_sec = time / 1000;
	      	timeout.tv_usec = (time % 1000) * 1000;
	      	ret = select (32, &fdmask, NULL, NULL, &timeout);
	      
	      	if (ret == -1)
	      	{
			  	if (errno!=EINTR)
			  		RFError("select error on events");
		  	} else if (ret == 0) {
		    	event->eventType = EVENT_TIMEOUT;
   			  	return;
		  	}

	  	}

      	if (FD_ISSET(socket, &fdmask)) 
      	{
	  		socklen_t fromLen = sizeof(event->eventSource);
	  		int cc;
	  		event->eventType = EVENT_NETWORK;
	  		event->eventDetail = new RF244BPacket;
	  		cc = recvfrom(socket, (char*)event->eventDetail, sizeof(RF244BPacket), 0,
		        (struct sockaddr *) &event->eventSource, &fromLen);
	 	 	if (cc <= 0)
	    	{
	      		if (cc < 0 && errno != EINTR)
					perror("event recvfrom");
	      		continue;
	    	}
	  		if (fromLen != sizeof(struct sockaddr_in))
	    		continue;
	  		ConvertIncoming(event->eventDetail);
	  		return;
		}

    }
}

/*
 * Resolve the specified host name into an internet address.  The "name" may
 * be either a character string name, or an address in the form a.b.c.d where
 * the pieces are octal, decimal, or hex numbers.  Returns a pointer to a
 * sockaddr_in struct (note this structure is statically allocated and must
 * be copied), or NULL if the name is unknown.
 */

Sockaddr *
resolveHost(register char *name)
{
	register struct hostent *fhost;
	struct in_addr fadd;
	static Sockaddr sa;

	if ((fhost = gethostbyname(name)) != NULL) {
		sa.sin_family = fhost->h_addrtype;
		sa.sin_port = 0;
		bcopy(fhost->h_addr, &sa.sin_addr, fhost->h_length);
	} else {
		fadd.s_addr = inet_addr(name);
		if (fadd.s_addr != -1) {
			sa.sin_family = AF_INET;	/* grot */
			sa.sin_port = 0;
			sa.sin_addr.s_addr = fadd.s_addr;
		} else
			return(NULL);
	}
	return(&sa);
}


/* Output error message */
void RFError(char *s)
{
	fprintf(stderr, "CS244B Distributed Replicated Files: %s\n", s);
	perror("CS244B Assignment 2");
	exit(-1);
}

/* Output warning message */
void RFWarn(char *s)
{
	printf("WARNING: Distributed Replicated Files: %s\n", s);
	return;
}

/* Output debug message */
void RFDebug(char *s)
{
	printf("Debug: %s\n", s);
	return;
}

/* Get current time */
int getCurrentTime () {
	struct timeval timenow;
	gettimeofday(&timenow, 0);
	return (timenow.tv_sec * 1000 + timenow.tv_usec / 1000 - TimeZero);

}

